#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2024/1/30 10:12
# @Author  : xinyingjie
# @File    : SendPreData.py
# @Description :
import threading
import time

from kafka import KafkaProducer, KafkaConsumer
import json
from config.config import KAFKA_HOST,AI_DATA_TOPIC,PRE_DATA_TOPIC,ALL_DATA_TOPIC


producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                             value_serializer=lambda x: json.dumps(x).encode('utf-8'))
# 指定文件夹路径
folder_path_pre_data = r'D:\工作\项目\浙江大桥\project\zhejaing_sumo\output-pre-data0201.txt'
folder_path_all_data = r'D:\工作\项目\浙江大桥\project\zhejaing_sumo\output-all-data0201.txt'
folder_path_ai_data = r'D:\工作\项目\浙江大桥\project\zhejaing_sumo\output-ai-data0201.txt'





def send_pre_data():
    time.sleep(4)
    with open(folder_path_pre_data, 'r', encoding='utf-8') as f:
        while True:
            line = f.readline().strip()
            if line:
                now_frame_data = json.loads(line)
                # print("pre-data",now_frame_data)
                producer.send(PRE_DATA_TOPIC, value=now_frame_data, key="key")
                time.sleep(now_frame_data["sleepTime"])
            else:
                break

def send_all_data():
    with open(folder_path_all_data, 'r', encoding='utf-8') as f:
        while True:
            line = f.readline().strip()
            if line:
                now_frame_data = json.loads(line)
                # print("all-data",now_frame_data)
                producer.send(ALL_DATA_TOPIC, value=now_frame_data, key="key")
                time.sleep(now_frame_data["sleepTime"])
            else:
                break


def send_ai_data():
    with open(folder_path_ai_data, 'r', encoding='utf-8') as f:
        while True:
            line = f.readline().strip()
            if line:
                now_frame_data = json.loads(line)
                print("ai-data",now_frame_data)
                producer.send(AI_DATA_TOPIC, value=now_frame_data, key="key")
                time.sleep(now_frame_data["sleepTime"])
            else:
                break


threading.Thread(target=send_pre_data).start()
threading.Thread(target=send_all_data).start()
threading.Thread(target=send_ai_data).start()